#include "stream/Reader.h"
#include "stream/Topic.h"

#include <iostream>

namespace stream {

Reader::Reader(std::shared_ptr<Topic> topic)
: _topic(topic) {
}

std::shared_ptr<Topic> Reader::topic() const{
	return _topic;
}

class MyReader : public Reader{
public:
	MyReader(std::shared_ptr<Topic> topic, size_t offset)
	: Reader(topic)
	, index(offset){
	}
	~MyReader() override {
        this->close();
    }
	
	bool hasNext() const override {
		return index < max;
	}
	Reader::record_type current() const override {
		return std::shared_ptr<std::string>(new std::string("1"));
	}
	void next() override{
		++index;
	}
protected:
	bool _open() override{
		return true;
	}
	void _close() override{
	}
private:
	size_t index = 0;
	size_t max = 3;
};

class TopicReader : public Reader {
public:
	TopicReader(std::shared_ptr<Topic> topic, size_t offset)
	: Reader(topic)
	, _offset(offset)
	{
	}
	~TopicReader() override{
		this->close();
	}

	bool hasNext() const override {
		return _offset < topic()->size();
	}
	Reader::record_type current() const override {
		const std::string index = std::to_string(_offset);
		std::shared_ptr<std::string> ret(new std::string);
		if(!topic()->get(index, *ret)){
			std::cerr << "get data failed from topic!" << std::endl;
		}
		return ret;
	}
	void next() override {
		//_offset += topic()->step();
		++_offset;
	}

protected:
	bool _open() override{
		bool ret = false;
		if(!topic()->isOpened()){
			ret = topic()->open();
		}
		else // already opened
			ret = true;

        size_t tsize = topic()->size();
        if(tsize > 0 && _offset >= tsize){
			_offset = tsize - 1;
		}
		return ret;
	}
	void _close() override{
		_offset = 0;
		if(topic()->isOpened()){
			if(topic().use_count() <= 1){
				topic()->close();
			}
		}
	}
private:
	size_t _offset = 0;
};

Reader * makeReaderMock(std::shared_ptr<Topic> topic, size_t offset){
    return new MyReader(topic, offset);
}
Reader * makeReader(std::shared_ptr<Topic> topic, size_t offset){
	return new TopicReader(topic, offset);
}

}//end namespace stream
